package com.atguigu.app.dws;

import com.atguigu.app.func.SplitFunction;
import com.atguigu.bean.KeywordStats;
import com.atguigu.uitls.ClickHouseUtil;
import com.atguigu.uitls.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//数据流：web/app -> nginx -> 日志服务器 -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse
//程  序：mock    -> nginx -> logger.sh -> Kafka(ZK) -> BaseLogApp -> Kafka -> KeywordStatsApp -> ClickHouse
public class KeywordStatsApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境与Kafka主题的分区数保持一致

        //CK
        //        env.setStateBackend(new FsStateBackend("hdfs://"));
        //        env.enableCheckpointing(5000L);
        //        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //        env.getCheckpointConfig().setCheckpointInterval(10000L);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.使用DDL方式 读取Kafka数据创建表,同时提取事件时间生成Watermark
        String groupId = "keyword_stats_app_210826";
        String pageViewSourceTopic = "dwd_page_log";

        tableEnv.executeSql("" +
                "create table page_log( " +
                "    page MAP<String,String>, " +
                "    ts bigint, " +
                "    rt as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000,'yyyy-MM-dd HH:mm:ss')), " +
                "    WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +
                ") with (" + MyKafkaUtil.getKafkaDDL(pageViewSourceTopic, groupId) + ")");

        //TODO 3.过滤出搜索数据
        Table keyWordTable = tableEnv.sqlQuery("" +
                "select " +
                "    page['item'] key_word, " +
                "    rt " +
                "from page_log " +
                "where page['last_page_id'] = 'search' " +
                "and page['item'] is not null");

        //TODO 4.注册UDTF函数
        tableEnv.createTemporarySystemFunction("SplitFunction", SplitFunction.class);

        //TODO 5.切词
        tableEnv.createTemporaryView("key_word_table", keyWordTable);
        Table wordTable = tableEnv.sqlQuery("" +
                "SELECT  " +
                "    rt, " +
                "    word  " +
                "FROM key_word_table, " +
                "LATERAL TABLE(SplitFunction(key_word))");

        //TODO 6.分组、开窗、聚合
        tableEnv.createTemporaryView("word_table", wordTable);
        Table resultTable = tableEnv.sqlQuery("" +
                "select " +
                "    'search' source, " +
                "    DATE_FORMAT(TUMBLE_START(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') stt, " +
                "    DATE_FORMAT(TUMBLE_end(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') edt, " +
                "    word keyword, " +
                "    count(*) ct, " +
                "    UNIX_TIMESTAMP()*1000 ts " +
                "from word_table " +
                "group by " +
                "    word, " +
                "    TUMBLE(rt, INTERVAL '10' second)");

        //TODO 7.将动态表转换为流
        DataStream<KeywordStats> keywordStatsDataStream = tableEnv.toAppendStream(resultTable, KeywordStats.class);

        //TODO 8.将数据写出到ClickHouse
        keywordStatsDataStream.print(">>>>>>>>>>>");
        keywordStatsDataStream.addSink(ClickHouseUtil.getSinkFunction("insert into keyword_stats_210826(word,ct,source,stt,edt,ts) values(?,?,?,?,?,?)"));

        //TODO 9.启动任务
        env.execute("KeywordStatsApp");

    }
}
